package com.csw.flink.sql

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

object Demo08MysqlToMysql {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()


    val bsEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, settings)

    bsEnv.executeSql(
      """
        |
        |CREATE TABLE mysql_student2 (
        |  id int,
        |  name string,
        |  age int ,
        |  gender string,
        |  clazz string
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://node1:3306/student?useUnicode=true&characterEncoding=utf-8&useSSL=false',
        |   'table-name' = 'student2',
        |   'username' = 'root',
        |   'password'= '123456'
        |
        |)
        |
      """.stripMargin)

    bsEnv.executeSql(
      """
        |
        |CREATE TABLE clazz_num (
        |  clazz STRING,
        |  num bigint,
        |  PRIMARY KEY (clazz) NOT ENFORCED
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/csw?useUnicode=true&characterEncoding=utf-8&useSSL=false',
        |   'table-name' = 'clazz-num',
        |   'username' = 'root',
        |   'password'= '123456'
        |
        |)
        |
      """.stripMargin)

    bsEnv.executeSql(
      """
        |
        |insert into clazz_num
        |select clazz,count(1) as num from mysql_student2 group by clazz
        |
      """.stripMargin)
  }
}
